package com.zxw.queue;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.zxw.entity.ObjectEvent;

import java.util.Iterator;
import java.util.List;
import java.util.Objects;


/**
 * @author by zhuxiansheng
 * @description Disrupter队列操作工具类
 * @date
 */

public class DisruptorQueue<T> {
    private Disruptor<ObjectEvent<T>> disruptor;
    private RingBuffer<ObjectEvent<T>> ringBuffer;


    public DisruptorQueue(Disruptor<ObjectEvent<T>> disruptor) {
        this.disruptor = disruptor;
        this.ringBuffer = disruptor.getRingBuffer();
        this.disruptor.start();
    }

    public void add(T o) {
        if (Objects.nonNull(o)) {
            long nextSeq = this.ringBuffer.next();

            try {
                ObjectEvent<T> event = this.ringBuffer.get(nextSeq);
                event.setEvent(o);
            } finally {
                this.ringBuffer.publish(nextSeq);
            }
        }
    }

    public void addAll(List<T> o) {
        if (o != null) {
            Iterator<T> iterator = o.iterator();
            while (iterator.hasNext()) {
                T next = iterator.next();
                if (next != null) {
                    this.add(next);
                }
            }
        }
    }

    public long cursor() {
        return this.disruptor.getRingBuffer().getCursor();
    }

    public void shutdown() {
        this.disruptor.shutdown();
    }

    public Disruptor<ObjectEvent<T>> getDisruptor() {
        return this.disruptor;
    }

    public void setDisruptor(Disruptor<ObjectEvent<T>> disruptor) {
        this.disruptor = disruptor;
    }

    public RingBuffer<ObjectEvent<T>> getRingBuffer() {
        return this.ringBuffer;
    }

    public void setRingBuffer(RingBuffer<ObjectEvent<T>> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

}
